package com.zhang.third.day09;

import com.zhang.third.utils.ClickEventSource;
import com.zhang.third.utils.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: FLink SQL
 * @author: zhang
 * @date: 2022/4/13 16:15
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<Event> stream = env
                .addSource(new ClickEventSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forMonotonousTimestamps()
                                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                                    @Override
                                    public long extractTimestamp(Event element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        // 将流转化为动态表
        Table table = tableEnv.fromDataStream(
                stream,
                $("key").as("username"),
                $("value").as("url"),
                // .rowtime表示ts是事件时间
                $("ts").rowtime().as("eventtime"),
                // 新添加一列处理时间，。.proctime表示这一列是处理时间 必须写在最后
                $("processingtime").proctime()
        );

        // 将动态表转化为流
        tableEnv.toChangelogStream(table).print();


        env.execute();
    }
}
